import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object TimeQuantumCntAnalysis {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("KeyAnalysis").setMaster("local[*]")
    val context = new SparkContext(conf)

    val line = context.textFile("d:\\SogouQ.txt")
    val dataArr: RDD[Array[String]] = line.map(_.split("\\s+"))
    val sougouDataLine: RDD[SogouRecord] = dataArr.map(arr => SogouRecord(arr(0), arr(1), arr(2), arr(3).toInt, arr(4).toInt, arr(5)))

    //TODO 统计每分钟的流量
    val minuteCntRDD: RDD[(String, Int)] = sougouDataLine.map {
      line => {
        val minute = line.queryTime.substring(0, 5)
        (minute, 1)
      }
    }
    val result: Array[(String, Int)] = minuteCntRDD.reduceByKey(_ + _).sortByKey().take(10)
    result.foreach(println)

    context.stop()
  }
  /**
   * 用户搜索点击网页记录Record
   * @param queryTime  访问时间，格式为：HH:mm:ss
   * @param userId     用户ID
   * @param queryWords 查询词
   * @param resultRank 该URL在返回结果中的排名
   * @param clickRank  用户点击的顺序号
   * @param clickUrl   用户点击的URL
   */
  case class SogouRecord(
                          queryTime: String,
                          userId: String,
                          queryWords: String,
                          resultRank: Int,
                          clickRank: Int,
                          clickUrl: String
                        )

}
